package com.llx.datastream







import java.sql.DriverManager

import java.sql.{Connection, PreparedStatement}
import org.apache.commons.configuration.Configuration
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.functions.source.RichSourceFunction
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object 自定义Source_mysql {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //通过env.addSource 添加输入源
    val source: DataStream[Student] = env.addSource(new SQL_source)
    source.print()
    env.execute()
  }


  class SQL_source extends RichSourceFunction[Student]{
    private var connection: Connection = null
    private var ps: PreparedStatement = null
//    override
     def open(parameters: Configuration): Unit = {
      val driver = "com.mysql.jdbc.Driver"
      val url = "jdbc:mysql://hadoop01:3306/test"
      val username = "root"
      val password = "root"
      Class.forName(driver)
      connection = DriverManager.getConnection(url, username, password)
      val sql = "select stuid , stuname , stuaddr , stusex from Student"
      ps = connection.prepareStatement(sql)
    }

    override def close(): Unit = {
      if(connection != null){
        connection.close()
      }
      if(ps != null){
        ps.close()
      }
    }

    override def run(sourceContext: SourceContext[Student]): Unit = {
      val queryRequest = ps.executeQuery()
      while (queryRequest.next()){
        val stuid = queryRequest.getInt("stuid")
        val stuname = queryRequest.getString("stuname")
        val stuaddr = queryRequest.getString("stuaddr")
        val stusex = queryRequest.getString("stusex")
        val stu = new Student(stuid , stuname , stuaddr , stusex)
        sourceContext.collect(stu)
      }
    }
    override def cancel(): Unit = {}
  }

  case class Student(stuid:Int , stuname:String , stuaddr:String , stusex:String){
    override def toString: String = {
      "stuid:"+stuid+" stuname:"+stuname+" stuaddr:"+stuaddr+" stusex:"+stusex
    }
  }
 
}
